package com.wish233.register.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.wish233.config.RegistryConfig;
import com.wish233.model.register.ServiceMetaInfo;
import com.wish233.register.Registry;
import com.wish233.utils.HeartBeatUtils;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;
import lombok.extern.slf4j.Slf4j;

import java.net.URI;
import java.time.Duration;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author WHH
 * 基于Redis的注册中心
 */
@Slf4j
public class RedisRegistry implements Registry {

    private static final String REDIS_ROOT_PATH = "wrpc:";
    private static final String REDIS_NODE_PATH = "wrpc:node:";
    private StatefulRedisConnection<String, String> connection;
    // 同步命令接口
    private RedisCommands<String, String> commands;
    private RedisClient client;
    private RegistryServiceCache cache;
    private RedisWatcher redisWatcher;


    @Override
    public void init(RegistryConfig registryConfig) {
        String password = registryConfig.getPassword();
        Integer database = registryConfig.getDatabase();
        Long timeout = registryConfig.getTimeout();
        URI uri = URI.create(registryConfig.getAddress());
        String host = uri.getHost();
        int port = uri.getPort() > 0 ? uri.getPort() : 6379;
        RedisURI redisUri = RedisURI.builder()
                .withHost(host)
                .withPort(port)
                .withPassword(password.toCharArray())
                .withTimeout(Duration.ofMillis(timeout))
                .withDatabase(database)
                .build();
        client = RedisClient.create(redisUri);
        connection = client.connect();
        commands = connection.sync();
        heartBeat();
        redisWatcher = new RedisWatcher(client);
        Runtime.getRuntime().addShutdownHook(new Thread(this::destroy));
        cache = RegistryServiceCache.getCache();
    }

    @Override
    public void register(ServiceMetaInfo serviceMetaInfo) throws Exception {
        String serviceKey = serviceMetaInfo.getServiceKey();
        String serviceNodeKey = serviceMetaInfo.getServiceNodeKey();
        // 1. 将节点信息存储到节点键
        String nodeKey = REDIS_NODE_PATH + serviceNodeKey;
        commands.set(nodeKey, JSONUtil.toJsonStr(serviceMetaInfo));
        commands.expire(nodeKey, Duration.ofMinutes(1));
        // 2. 将节点引用添加到服务列表
        String serviceListKey = REDIS_ROOT_PATH + serviceKey;
        commands.lpush(serviceListKey, serviceNodeKey);
        commands.expire(serviceListKey, Duration.ofMinutes(1));
        //心跳检测
        HeartBeatUtils.addNode(nodeKey);
        HeartBeatUtils.addNode(serviceListKey);
    }

    @Override
    public void unRegister(ServiceMetaInfo serviceMetaInfo) {
        String serviceKey = serviceMetaInfo.getServiceKey();
        String serviceNodeKey = serviceMetaInfo.getServiceNodeKey();
        // 1. 从服务列表中移除节点引用
        String serviceListKey = REDIS_ROOT_PATH + serviceKey;
        commands.lrem(serviceListKey, 0, serviceNodeKey);
        // 2. 删除节点数据
        String nodeKey = REDIS_NODE_PATH + serviceNodeKey;
        commands.del(nodeKey);
        //心跳检测
        HeartBeatUtils.removeNode(nodeKey);
        HeartBeatUtils.removeNode(serviceListKey);
        log.info("服务节点 {} 已下线", serviceNodeKey);
    }

    @Override
    public List<ServiceMetaInfo> serviceDiscovery(String serviceKey) {
        //读缓存
        List<ServiceMetaInfo> serviceMetaInfos = cache.readCache(serviceKey);
        if (CollectionUtil.isNotEmpty(serviceMetaInfos)) {
            return serviceMetaInfos;
        }
        String serviceListKey = REDIS_ROOT_PATH + serviceKey;
        // 1. 获取该服务的所有节点引用
        List<String> nodeKeys = commands.lrange(serviceListKey, 0, -1);
        // 2. 批量获取节点详情
        List<ServiceMetaInfo> metaInfoList = nodeKeys.stream()
                .map(nodeKey -> {
                    String info = commands.get(REDIS_NODE_PATH + nodeKey);
                    watch(nodeKey);
                    return info;
                })
                .filter(Objects::nonNull)
                .map(json -> JSONUtil.toBean(json, ServiceMetaInfo.class))
                .collect(Collectors.toList());
        cache.writeCache(serviceKey, metaInfoList);
        return metaInfoList;
    }

    @Override
    public void heartBeat() {
        HeartBeatUtils.doHeartBeat(key -> {
            //给每个key续期
            commands.expire(key, Duration.ofSeconds(30));
            log.info("heartBeat {}", key);
        });
    }

    @Override
    public void destroy() {
        Set<String> set = HeartBeatUtils.getSet();
        //下线所有节点
        for (String key : set) {
            commands.del(key);
        }
        log.info("节点下线！");
        Optional.of(connection).ifPresent(StatefulRedisConnection::close);
        Optional.of(client).ifPresent(RedisClient::shutdown);
    }

    @Override
    public void watch(String serviceNodeKey) {
        redisWatcher.watch(serviceNodeKey);
    }
}
